package io.kimmking.javacourse.mq.activemq;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQTopic;

import javax.jms.*;
import java.util.concurrent.atomic.AtomicInteger;


public class ActivemqConsumer {

    public static void main(String[] args) {

        /**
         * 两种Destination
         * 1。 queue  点对点模式
         * 2。 topic  发布订阅的模式
         * 定义Destination
         */

        Destination destination = new ActiveMQTopic("test.topic");
//        Destination destination = new ActiveMQQueue("test.queue");

        consumer(destination);

    }

    public static void consumer(Destination destination) {
        try {
            // 创建连接和会话
            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.1.59:61616");
            ActiveMQConnection conn = (ActiveMQConnection) factory.createConnection();
            conn.start();
            Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 创建消费者
            MessageConsumer consumer = session.createConsumer( destination );
            final AtomicInteger count = new AtomicInteger(0);
            MessageListener listener = message -> {
                try {
                    // 打印所有的消息内容
                    // Thread.sleep();
                    System.out.println(count.incrementAndGet() + " => receive from " + destination.toString() + ": " + message);
                    // message.acknowledge(); // 前面所有未被确认的消息全部都确认。

                } catch (Exception e) {
                    e.printStackTrace(); // 不要吞任何这里的异常，
                }
            };
            // 绑定消息监听器
            consumer.setMessageListener(listener);


            /**
             * 也可以使用receive的方式来做
             */
            //consumer.receive()

//            Thread.sleep(20000);
//            session.close();
//            conn.close();



        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
